#include "Connection.h"
#include <EventLoop.h>

Connection::Connection(EventLoop *loop, int fd, const std::string &name)
    : loop_(loop),
      state_(kConnecting),
      channel_(new Channel(loop, fd)),
      socket_(new Socket(fd)),
      name_(name),
      highWaterMark_(64 * 1024 * 1024),
      reading_(true)
{
    channel_->setReadCallback(std::bind(&Connection::handleRead, this));
    channel_->setWriteCallback(std::bind(&Connection::handleWrite, this));
    channel_->setCloseCallback(std::bind(&Connection::handleClose, this));
    channel_->setErrorCallback(std::bind(&Connection::handleError, this));
    LOG_DEBUG << "TcpConnection::ctor[" << name_ << "] at " << this
              << " fd=" << fd;
    socket_->setKeepAlive(true);
}

std::string Connection::stateToString()
{
    switch (state_)
    {
    case kConnecting:
        return "connecting";
    case kConnected:
        return "connected";
    case kDisconnecting:
        return "disConnecting";
    case kDisconnected:
        return "disConnected";
    default:
        return "unknown state";
    }
}

void Connection::setTcpNoDelay(bool on)
{
    socket_->setTcpNoDelay(on);
}

void Connection::handleRead()
{
    loop_->assertInLoopThread();
    int saveError = 0;
    ssize_t n = outputBuffer_.readFd(channel_->fd(), &saveError);
    if (n > 0)
    {
        messageCallback_(shared_from_this(), &inputBuffer_);
    }
    else if (n == 0)
    {
        handleClose();
    }
    else
    {
        errno = saveError;
        LOG_ERROR << "TcpConnection::handleRead";
        handleError();
    }
}

void Connection::handleWrite()
{
    loop_->assertInLoopThread();
    if (channel_->isWriting())
    {
        ssize_t n = sockets::write(channel_->fd(), outputBuffer_.peek(), outputBuffer_.readableBytes());
        if (n > 0)
        {
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            {
                channel_->disableWrite();
                if (writeCompleteCallback_)
                {
                    loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
                }
                if (state_ == kDisconnecting)
                {
                    shutdownInLoop();
                }
            }
        }
        else
        {
            LOG_ERROR << "TcpConnection::handleWrite";
        }
    }
    else
    {
        LOG_TRACE << "Connection fd = " << channel_->fd()
                  << " is down, no more writing";
    }
}

void Connection::handleClose()
{
    loop_->assertInLoopThread();
    LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
    assert(state_ == kConnected || state_ == kDisconnecting);
    setState(kDisconnected);
    channel_->disableAll();
    // we don't close fd, leave it to dtor, so we can find leaks easily.
    TcpConnectionPtr guardThis(shared_from_this());
    connectionCallback_(guardThis);
    // must be the last line
    closeCallback_(guardThis);
}

void Connection::handleError()
{
    int err = sockets::getSocketError(channel_->fd());
    LOG_ERROR << "TcpConnection::handleError [" << name_
              << "] - SO_ERROR = " << err << " " << strerror(err);
}

void Connection::startReadInLoop()
{
    loop_->assertInLoopThread();
    if (!reading_ || !channel_->isReading())
    {
        channel_->enableRead();
        reading_ = true;
    }
}

void Connection::stopReadInLoop()
{
    loop_->assertInLoopThread();
    if (reading_ || channel_->isReading())
    {
        channel_->disableRead();
        reading_ = false;
    }
}

void Connection::sendInLoop(const std::string &msg)
{
    sendInLoop(msg.c_str(), msg.size());
}

void Connection::sendInLoop(const char *msg, size_t len)
{
    loop_->assertInLoopThread();
    if (state_ == kDisconnected)
    {
        LOG_WARN << "disconnected, give up writing";
        return;
    }
    ssize_t nWrite = 0;
    bool faultError = false;
    ssize_t remaining = len;
    // 如果outputbuffer中为空，可以尝试直接写
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    {
        nWrite = sockets::write(channel_->fd(), msg, len);
        if (nWrite >= 0)
        {
            remaining -= nWrite;
            if (remaining == 0 && writeCompleteCallback_)
            {
                loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
            }
        }
        else
        {
            nWrite = 0;
            if (errno != EWOULDBLOCK)
            {
                LOG_ERROR << "TcpConnection::sendInLoop";
                if (errno == EPIPE || errno == ECONNRESET) // why?
                {
                    faultError = true;
                }
            }
        }
    }

    assert(remaining <= len);
    if (remaining > 0 && !faultError)
    {
        size_t oldLen = outputBuffer_.readableBytes();
        if (oldLen + remaining >= highWaterMark_ &&
            oldLen < highWaterMark_ &&
            highWaterMarkCallback_)
        {
            loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
        }
        outputBuffer_.append(static_cast<const char *>(msg) + nWrite, remaining);
        if (!channel_->isWriting())
        {
            channel_->enableWrite();
        }
    }
}

void Connection::connectEstablished()
{
    loop_->assertInLoopThread();
    assert(state_ == kConnecting);
    setState(kConnected);
    channel_->tie(shared_from_this());
    channel_->enableRead();
    connectionCallback_(shared_from_this());
}

void Connection::connectDestroyed()
{
    loop_->assertInLoopThread();
    if (state_ == kConnected)
    {
        setState(kDisconnected);
        channel_->disableAll();
        connectionCallback_(shared_from_this()); // 通知上层应用层连接状态改变
    }
    channel_->remove();
}

void Connection::send(const std::string &msg)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread())
        {
            sendInLoop(msg);
        }
        else
        {
            using Func = void (Connection::*)(const std::string &msg);
            Func callback = &Connection::sendInLoop;
            loop_->runInLoop(std::bind(callback, this, msg));
        }
    }
}

void Connection::send(const char *msg, size_t len)
{
    sendInLoop(std::string(msg, len));
}

void Connection::shutdown()
{
    if (state_ == kConnected)
    {
        setState(kDisconnecting);
        loop_->runInLoop(std::bind(&Connection::shutdownInLoop, this));
    }
}

void Connection::shutdownInLoop()
{
    loop_->assertInLoopThread();
    if (!channel_->isWriting()) // 如果正在写，要等待写完成保证数据的完整发送
    {
        socket_->shutdownWrite();
    }
}

Connection::~Connection()
{
    LOG_DEBUG << "TcpConnection::dtor[" << name_ << "] at " << this
              << " fd=" << channel_->fd()
              << " state=" << stateToString();
    assert(state_ == kDisconnected);
}
